package com.mxd.flink.connector.utils;

import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.binary.BinaryStringData;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Base64;

/**
 * 这里时redis行数据转换工具，直接抄的网上的
 */
public class RedisRowUtil {

    public static Object dataTypeFromString(LogicalType fieldType, String result) {
        if (result == null) {
            return null;
        }
        return createDeserializer(fieldType).deserialize(result);
    }

    public static String rowDataToString(LogicalType fieldType, RowData rowData, Integer index) {
        if (rowData.isNullAt(index.intValue())) {
            return null;
        }
        return createSerializer(fieldType).serialize(rowData, index);
    }

    public static RedisDeserializationConverter createDeserializer(LogicalType fieldType) {
        switch (fieldType.getTypeRoot()) {
            case BIGINT:
            case INTERVAL_DAY_TIME: {
                return Long::valueOf;
            }
            case FLOAT: {
                return Float::valueOf;
            }
            case DOUBLE: {
                return Double::valueOf;
            }
            case CHAR:
            case VARCHAR: {
                return BinaryStringData::fromString;
            }
            case BOOLEAN: {
                return Boolean::valueOf;
            }
            case BINARY:
            case VARBINARY: {
                return result -> Base64.getDecoder().decode(result);
            }
            case DECIMAL: {
                DecimalType decimalType = (DecimalType)fieldType;
                int precision = decimalType.getPrecision();
                int scale = decimalType.getScale();
                return result -> {
                    BigDecimal decimal = new BigDecimal(result);
                    return DecimalData.fromBigDecimal((BigDecimal)decimal, (int)precision, (int)scale);
                };
            }
            case TINYINT: {
                return Byte::valueOf;
            }
            case SMALLINT: {
                return Short::valueOf;
            }
            case INTEGER:
            case DATE:
            case INTERVAL_YEAR_MONTH: {
                return Integer::valueOf;
            }
            case TIME_WITHOUT_TIME_ZONE: {
                int precision = LogicalTypeChecks.getPrecision((LogicalType)fieldType);
                if (precision < 0 || precision > 3) {
                    throw new UnsupportedOperationException(String.format("The precision %s of Time type is out of range [%s, %s]", precision, 0, 3));
                }
                return Integer::valueOf;
            }
            case TIMESTAMP_WITHOUT_TIME_ZONE:
            case TIMESTAMP_WITH_LOCAL_TIME_ZONE: {
                int precision = LogicalTypeChecks.getPrecision((LogicalType)fieldType);
                if (precision < 0 || precision > 3) {
                    throw new UnsupportedOperationException(String.format("The precision %s of Timestamp is out of range [%s, %s]", precision, 0, 3));
                }
                return result -> {
                    long milliseconds = Long.valueOf(result);
                    return TimestampData.fromEpochMillis((long)milliseconds);
                };
            }
        }
        throw new UnsupportedOperationException("Unsupported field type: " + fieldType);
    }

    private static RedisSerializationConverter createSerializer(LogicalType fieldType) {
        switch (fieldType.getTypeRoot()) {
            case CHAR:
            case VARCHAR: {
                return (rowData, index) -> rowData.getString(index.intValue()).toString();
            }
            case BOOLEAN: {
                return (rowData, index) -> String.valueOf(rowData.getBoolean(index.intValue()));
            }
            case BINARY:
            case VARBINARY: {
                return (rowData, index) -> Base64.getEncoder().encodeToString(rowData.getBinary(index.intValue()));
            }
            case DECIMAL: {
                DecimalType decimalType = (DecimalType)fieldType;
                int precision = decimalType.getPrecision();
                int scale = decimalType.getScale();
                return (rowData, index) -> {
                    BigDecimal decimal = rowData.getDecimal(index.intValue(), precision, scale).toBigDecimal();
                    return decimal.toString();
                };
            }
            case TINYINT: {
                return (rowData, index) -> String.valueOf(rowData.getByte(index.intValue()));
            }
            case SMALLINT: {
                return (rowData, index) -> String.valueOf(rowData.getShort(index.intValue()));
            }
            case INTEGER:
            case DATE:
            case INTERVAL_YEAR_MONTH: {
                return (rowData, index) -> String.valueOf(rowData.getInt(index.intValue()));
            }
            case TIME_WITHOUT_TIME_ZONE: {
                int precision = LogicalTypeChecks.getPrecision((LogicalType)fieldType);
                if (precision < 0 || precision > 3) {
                    throw new UnsupportedOperationException(String.format("The precision %s of Time type is out of range [%s, %s]", precision, 0, 3));
                }
                return (rowData, index) -> String.valueOf(rowData.getInt(index.intValue()));
            }
            case BIGINT:
            case INTERVAL_DAY_TIME: {
                return (rowData, index) -> String.valueOf(rowData.getLong(index.intValue()));
            }
            case FLOAT: {
                return (rowData, index) -> String.valueOf(rowData.getFloat(index.intValue()));
            }
            case DOUBLE: {
                return (rowData, index) -> String.valueOf(rowData.getDouble(index.intValue()));
            }
            case TIMESTAMP_WITHOUT_TIME_ZONE:
            case TIMESTAMP_WITH_LOCAL_TIME_ZONE: {
                int precision = LogicalTypeChecks.getPrecision((LogicalType)fieldType);
                if (precision < 0 || precision > 3) {
                    throw new UnsupportedOperationException(String.format("The precision %s of Timestamp is out of range [%s, %s]", precision, 0, 3));
                }
                return (rowData, index) -> String.valueOf(rowData.getTimestamp(index.intValue(), precision).getMillisecond());
            }
        }
        throw new UnsupportedOperationException("Unsupported field type: " + fieldType);
    }

    @FunctionalInterface
    static interface RedisSerializationConverter
            extends Serializable {
        public String serialize(RowData var1, Integer var2);
    }

    @FunctionalInterface
    static interface RedisDeserializationConverter
            extends Serializable {
        public Object deserialize(String var1);
    }
}